"""
Case Type   : HTAP行列融合特性
Case Name   : 行列转换过程中节点异常(kill gsql进程)
Create At   : 2025/2/11
Owner       : ningyali
Description :
    1.建表插入数据
    2.开启session1，等待1s后线程执行kill gsql进程
    3.进行全表转换
    4.查看query_imcstore_views()
    5.查询
    6.重启数据库
    7.查看query_imcstore_views()
    8.查询
    9.进行全表转换
    10.查看query_imcstore_views()
    11.查询
    12.清理数据环境
Expect      :
    1.执行成功
    2.执行成功
    3.执行失败
    4.杀进程不影响转换，转换成功
    5.执行成功
    6.执行成功
    7.执行成功，查询结果为空
    8.执行成功，查询计划不走imcs scan
    9.执行成功
    10.执行成功，查询结果不为空
    11.执行成功，查询计划走imcs scan
    12.执行成功
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread
from testcase.utils.Logger import Logger
from yat.test import macro


class HTAP(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('SingleDbUser')
        self.Constant = Constant()
        self.common = Common()
        self.t_name = 't_htap_single_004'
        self.clean_table = f'drop table if exists {self.t_name};'
        self.q_view_sql = f"select relname,imcs_attrs,imcs_nattrs," \
                          f"imcs_status,is_partition " \
                          f"from query_imcstore_views() " \
                          f"where relname like '{self.t_name}%' " \
                          f"order by relname;"
        self.q_data_sql = f'set enable_imcsscan to on;' \
                          f'explain (costs false) ' \
                          f'select count(c1) from {self.t_name};' \
                          f'select count(c1) from {self.t_name};'

    def test_trans(self):
        text = '----step1:建表插入数据; expect:执行成功----'
        self.log.info(text)
        sql = f'''{self.clean_table}
            create table {self.t_name}(
                c1 integer(20),
                c2 varchar(20) not null default '',
                c3 char(30)
            )partition by range (c1)(
                partition {self.t_name}_p1 values less than (1000),
                partition {self.t_name}_p2 values less than (2000),
                partition {self.t_name}_p3 values less than (3000),
                partition {self.t_name}_p4 values less than (maxvalue));
            insert into {self.t_name} values (
            generate_series(1, 10000000),
            generate_series(1, 10000000)||'aaa',
            generate_series(1, 10000000)||'bbb');'''
        sql_res = self.pri_sh.executDbSql(
            sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertIn('CREATE TABLE', sql_res, f'执行失败:{text}')
        self.assertIn('INSERT 0 10000000', sql_res, f'执行失败:{text}')

        text = '----step2:开启session1，等待1s后线程执行ctrl+c进程; expect:执行成功----'
        self.log.info(text)
        cmd = "sleep 1;" \
              "ps ux|grep gsql|grep imcstored|awk '{{print $2}}'|" \
              "xargs -I {{}} kill -9 {{}}"
        kill_thread = ComThread(self.pri_sh.node.sh, args=(cmd,))
        kill_thread.setDaemon(True)
        kill_thread.start()

        text = '----step3:行列转换; expect:执行失败----'
        self.log.info(text)
        alter_sql = f'alter table {self.t_name} imcstored;'
        sql_res = self.pri_sh.executDbSql(
            alter_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertNotIn('ALTER TABLE', sql_res, f'执行失败:{text}')

        kill_thread.join(300)
        kill_res = kill_thread.get_result().result()
        self.log.info(kill_res)
        self.assertEqual(len(kill_res.strip()), 0, f'执行失败:{text}')

        text = '----step4:查看query_imcstore_views; expect:杀进程不影响转换，转换成功----'
        self.log.info(text)
        self.log.info('转换时间不确定，循环等待以减少不必要的等待')
        sql_res = None
        for i in range(10):
            sql_res = self.pri_sh.executDbSql(
                self.q_view_sql, env_path=macro.SINGLE_DB_ENV_PATH)
            if sql_res.count('COMPLETE') == 5:
                break
            time.sleep(3)
        self.assertIn('5 rows', sql_res, f'执行失败:{text}')
        self.assertEqual(sql_res.count('COMPLETE'), 5, f'执行失败:{text}')

        text = '----step5:查询; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.q_data_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertIn('IMCStore Scan', sql_res, f'执行失败:{text}')
        self.assertIn('10000000', sql_res, f'执行失败:{text}')

        text = '----step6:重启数据库; expect:执行成功----'
        self.log.info(text)
        restart_res = self.pri_sh.om_restart(macro.SINGLE_DB_ENV_PATH)
        self.assertTrue(restart_res, f'执行失败:{text}')

        text = '----step7:查看query_imcstore_views; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.q_view_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertIn('0 rows', sql_res, f'执行失败:{text}')

        text = '----step8:查询; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.q_data_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertNotIn('IMCStore Scan', sql_res, f'执行失败:{text}')
        self.assertIn('10000000', sql_res, f'执行失败:{text}')

        text = '----step9:行列转换; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            alter_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertEqual('ALTER TABLE', sql_res.strip(), f'执行失败:{text}')

        text = '----step10:查看query_imcstore_views; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.q_view_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertEqual(sql_res.count('COMPLETE'), 5, f'执行失败:{text}')
        self.assertIn('5 rows', sql_res, f'执行失败:{text}')

        text = '----step11:查询; expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.q_data_sql, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertIn('IMCStore Scan', sql_res, f'执行失败:{text}')
        self.assertIn('10000000', sql_res, f'执行失败:{text}')

    def tearDown(self):
        text = '----step12:清理环境;expect:执行成功----'
        self.log.info(text)
        sql_res = self.pri_sh.executDbSql(
            self.clean_table, env_path=macro.SINGLE_DB_ENV_PATH)
        self.assertIn('DROP TABLE', sql_res, f'执行失败:{text}')

        self.log.info(f'-----{os.path.basename(__file__)} end-----')
